package com.atguigu.tingshu.album.canal;

import cn.hutool.core.lang.UUID;
import com.alibaba.fastjson.JSON;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.atguigu.tingshu.common.constant.KafkaConstant;
import com.atguigu.tingshu.common.constant.RedisConstant;
import com.atguigu.tingshu.common.service.KafkaService;
import com.atguigu.tingshu.vo.album.TrackStatMqVo;
import com.xpand.starter.canal.annotation.CanalEventListener;
import com.xpand.starter.canal.annotation.ListenPoint;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;

import java.util.List;
import java.util.Set;

@Slf4j
@CanalEventListener
public class CanalClientListener {

    @Autowired
    private RedisTemplate redisTemplate;
    @Autowired
    private KafkaService kafkaService;

    @ListenPoint(destination = "example",schema = "tingshu_album",table = "album_info",eventType = {CanalEntry.EventType.UPDATE })
    public void listenerAlbumInfo(CanalEntry.RowData rowData ){
        //获取修改之前的数据
//        List<CanalEntry.Column> beforeColumnsList = rowData.getBeforeColumnsList();//获取修改之前的数据集合
//        beforeColumnsList.forEach(column -> {
//            String columnName = column.getName();//获取列名
//            String columnValue = column.getValue();//获取列值
//            log.info("columnName:{},columnValue:{}",columnName,columnValue);
//        });
        System.out.println("============================================");
        //获取修改之后的数据
        List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
        afterColumnsList.forEach(column -> {
            String columnName = column.getName();//获取列名

            if (("id").equals(columnName)){
                String columnValue = column.getValue();//获取列值
                redisTemplate.delete(RedisConstant.ALBUM_INFO_PREFIX+columnValue);
                log.info("从Redis中删除了相关数据.......专辑id:{}",columnValue);
            }

        });

    }


    @ListenPoint(destination = "example",schema = "tingshu_album",table = "album_stat",eventType = {CanalEntry.EventType.UPDATE })
    public void listenerAlbumStat(CanalEntry.RowData rowData){
        //获取修改之后的数据
        List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
        Long albumId = null ;
        String statType = null;
        Integer count = null;
        for(CanalEntry.Column column : afterColumnsList) {

            String columnName = column.getName();
            // 判断列是否为专辑的id
            if("album_id".equals(columnName)) {
                String columnValue = column.getValue();     // 专辑的id
                redisTemplate.delete("album:stat:" + columnValue);
                log.info("从Redis中删除了专辑的统计数据，albumId: {}" , columnValue);
                albumId = Long.parseLong(columnValue);
            }

            if("stat_type".equals(columnName)) {
                statType = column.getValue();       // 获取统计维度
            }

            if("stat_num".equals(columnName)) {
                count = Integer.parseInt(column.getValue());
            }
        }
        //更新ES中专辑统计数据,使用MQ实现数据同步
        TrackStatMqVo trackStatMqVo = new TrackStatMqVo();
        trackStatMqVo.setBusinessNo(UUID.randomUUID().toString().replace("-",""));
        trackStatMqVo.setAlbumId(albumId);
        trackStatMqVo.setStatType(statType);
        trackStatMqVo.setCount(count);

        kafkaService.sendMessage(KafkaConstant.QUEUE_ALBUM_STAT_UPDATE, JSON.toJSONString(trackStatMqVo));
        log.info("发送同步消息到Kafka中.........");
    }

    // 同步更新用户数据
    @ListenPoint(destination = "example" , schema = "tingshu_user" , table = "user_info" , eventType = {CanalEntry.EventType.UPDATE } )
    public void listenerUserInfo(CanalEntry.RowData rowData) {

        // 获取修改之后的数据
        List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
        afterColumnsList.forEach(column -> {
            String columnName = column.getName();
            if("id".equals(columnName)) {
                String columnValue = column.getValue();     // 获取用户的id
                redisTemplate.delete("user:info:" + columnValue);
                log.info("删除了Redis中的用户数据...userId: {}" , columnValue);
            }
        });

    }

    // 分类数据
    @ListenPoint(destination = "example" , schema = "tingshu_album" , table = { "base_category1" , "base_category2" , "base_category3" } , eventType = {CanalEntry.EventType.UPDATE } )
    public void listenerCategory(CanalEntry.RowData rowData) {
        //删除分类的缓存数据
        Set keys = redisTemplate.keys("album:category:view:*");
        keys.forEach(key ->redisTemplate.delete(key));
        //从Redis中删除了分类数据
    }


}
